package part03;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/*
 * 计算数据流平均值
 */
public class StateDemo {
    public static void main(String[] args) throws Exception {
        //1，获取flink Stream执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.enableCheckpointing(2000);
        //2，获取数据流
        DataStreamSource<String> data = env.socketTextStream("centos7-3", 9999);
        //3,处理数据
        SingleOutputStreamOperator<Tuple2<String, String>> map = data.map(new MapFunction<String, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> map(String s) throws Exception {
                System.out.println(s);
                return new Tuple2<>(String.valueOf(1), s);  //转换格式
            }
        });
        //4，聚合操作
        KeyedStream<Tuple2<String, String>, String> keyBy = map.keyBy(f -> f.f0);
        //5，使用状态机制计算流式数据平均值
        SingleOutputStreamOperator<Tuple2<Long, Long>> average = keyBy.flatMap(new RichFlatMapFunction<Tuple2<String, String>, Tuple2<Long, Long>>() {
            ValueState<Tuple2<Long, Long>> sumState;

            @Override
            public void open(Configuration parameters) throws Exception {
                //创建state
                ValueStateDescriptor<Tuple2<Long, Long>> stateDescriptor = new ValueStateDescriptor<Tuple2<Long, Long>>(
                        "average",
                        TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {
                        }),
                        Tuple2.of(0L, 0L)
                );
                sumState = getRuntimeContext().getState(stateDescriptor);
                super.open(parameters);
            }

            @Override
            public void flatMap(Tuple2<String, String> stringStringTuple2, Collector<Tuple2<Long, Long>> collector) throws Exception {
                //获取state的值
                Tuple2<Long, Long> tmpValue = sumState.value();
                //数据计算
                tmpValue.f0 += 1;
                tmpValue.f1 += Long.parseLong(stringStringTuple2.f1);
                //更新state的值
                sumState.update(tmpValue);
                //计算平均值
                long average = tmpValue.f1 / tmpValue.f0;
                collector.collect(new Tuple2<>(Long.parseLong(stringStringTuple2.f0), average));
                //清空状态值
//                sumState.clear();
            }
        });
        //数据输出
        average.print();
        //触发作业
        env.execute();
    }
}
